import logging
import re
from urllib.parse import urlparse

import requests
from bs4 import BeautifulSoup
from flask import Blueprint, redirect
from flask import Response
from flask import jsonify
from flask import request
from sqlalchemy.orm.session import Session

from airflow.patsnap.service import task_service, alert_service
from airflow.patsnap.util.common_util import CommonUtil
from airflow.patsnap.util.etl_util import EtlUtil
from airflow.configuration import conf
from airflow.models.idata_task import Task, TaskType
from airflow.models.idata_workflow import WorkFlow
from airflow.models.taskinstance import TaskInstance
from airflow.utils import dates
from airflow.utils.session import provide_session
from airflow.utils.state import State
from airflow.www.api import provide_user, exception_handle

log = logging.getLogger(__name__)

task_bp = Blueprint('task_bp', __name__)


@task_bp.route('/health', methods=['GET'])
@exception_handle
def health():
    return jsonify(code=0, message='ok', data=None)


@task_bp.route('/tasks/<int:task_id>/trigger', methods=['POST'])
@exception_handle
@provide_session
def task_run(task_id, session: Session = None):
    import time
    execution_date = None
    data = request.get_json()
    if data:
        execution_date = data.get('execution_date', None)
        if execution_date:
            ts = int(execution_date)
            if ts > int(time.time() * 1000):
                return jsonify(code=400, message='计划时间不能选择未来时间', data=None)
    task = Task.get_task(task_id, session=session)
    if task:
        if task.task_concurrency:
            running = session.query(TaskInstance).filter(TaskInstance.task_id == 'task-{}'.format(task_id),
                                                         TaskInstance.state == State.RUNNING).all()
            if len(running) >= task.task_concurrency:
                return jsonify(code=1, message='task_concurrency exceed {}'.format(task.task_concurrency),
                               data=None)
            if len(running) >= 1 and task.type == 1:
                return jsonify(code=1, message='Streaming 任务不能多次运行',
                               data=None)
        resp = task_service.run_by_scheduler(task, execution_date, session=session)
        return jsonify(code=0, message='ok', data=resp)
    else:
        return jsonify(code=404, message='task not found 请确认是否保存', data=None)


@task_bp.route('/tasks/type', methods=['GET'])
@exception_handle
def get_task_types():
    options = [{'value': 'SHELL', 'label': 'Shell'}, {'value': 'SPARK', 'label': 'Spark'},
               {'value': 'FLINK', 'label': 'Flink'}, {'value': 'LOGSTASH', 'label': 'Logstash'},
               {'value': 'DAGRUN', 'label': 'DagRun'}, {'value': 'DATAFACTORY', 'label': 'DataFactory'}]
    return jsonify(code=0, message='ok', data=options)


@task_bp.route('/tasks/workflow_instance', methods=['GET'])
@exception_handle
def get_task_of_workflow_instance():
    dag_id = request.args.get('dag_id', None)
    execution_date = request.args.get('execution_date', None)
    if dag_id and execution_date:
        date = dates.from_milliseconds(execution_date)
        rs = task_service.tasks_of_workflow_instance(dag_id=dag_id, execution_date=date)
        return jsonify(code=0, message='ok', data=rs)
    else:
        return jsonify(code=400, message='dag_id or execution_date 不能为空', data=None)


@task_bp.route('/tasks/<int:task_id>', methods=['GET'])
@exception_handle
def task_get(task_id):
    task = Task.get_task(task_id)
    if task:
        # 查询taskinstance
        _task_instance = None
        if task.type == TaskType.STREAM:
            task_instance = TaskInstance.get_by_id(task_id)
            if task_instance:
                _task_instance = task_instance.to_json()
        return jsonify(code=0, message='ok', data=task.to_json(task_instance=_task_instance))
    else:
        return jsonify(code=404, message='task not found', data=None)


@task_bp.route('/tasks/stream', methods=['GET'])
@exception_handle
def get_stream_tasks():
    project_id = request.args.get('project_id', None)
    name = request.args.get('name', None)
    page = request.args.get('page', 0, type=int)
    operator = request.args.get('operator', None, type=str)
    result = task_service.get_stream_tasks(project_id, name, operator, page)
    return jsonify(code=0, message='ok', data=result)


@task_bp.route('/tasks', methods=['POST'])
@provide_user
@exception_handle
def task_insert(oa_user_name):
    data = request.get_json()
    owner = data.get('owner', None)
    if owner is None and oa_user_name:
        data['owner'] = oa_user_name
    task = Task(**data)
    task_service.insert_task(task)
    return jsonify(code=0, message='ok', data=task.to_json())


@task_bp.route('/tasks/<int:task_id>', methods=['PATCH'])
@exception_handle
def task_patch(task_id):
    '''
    Task 局部更新
    name folder_id operator content param files owner email_alert retries type
    :return:
    '''
    data = request.get_json()
    error = task_service.task_validate(task_id, data)
    if error is None:
        task_service.patch_task(task_id, data)
        return jsonify(code=0, message='ok', data=None)
    else:
        return jsonify(code=400, message=error, data=None)


@task_bp.route('/tasks/<int:task_id>/clone', methods=['POST'])
@exception_handle
def task_clone(task_id):
    '''
    Task 克隆
    :return:
    '''
    task = Task.get_task(task_id)
    new_task = Task(project_id=task.project_id, folder_id=task.folder_id, name=task.name + '_copy',
                    description=task.description,
                    operator=task.operator, content=task.content, param=task.param, env=task.env, files=task.files,
                    pool=task.pool, owner=task.owner, email_alert=task.email_alert, email=task.email,
                    version=task.version,
                    retries=task.retries, task_concurrency=task.task_concurrency,
                    execution_timeout=task.execution_timeout,
                    type=task.type, blood=task.blood)
    if len(new_task.name) > 250:
        return jsonify(code=500, message='task name is too long, must less than 250', data=None)
    task_service.insert_clone_task(new_task)
    return jsonify(code=0, message='ok', data=new_task.to_json())


@task_bp.route('/tasks/<int:task_id>', methods=['DELETE'])
@exception_handle
def task_delete(task_id):
    msg = task_service.delete_task(task_id)
    if msg:
        return jsonify(code=400, message=msg, data=None)
    else:
        return jsonify(code=0, message='ok', data=None)


@task_bp.route('/tasks', methods=['GET'])
@exception_handle
def task_query():
    project_id = request.args.get('project_id', None)
    folder_id = request.args.get('folder_id', None)
    if project_id is None and folder_id is None:
        return jsonify(code=1, message='project_id or folder_id 不能为空', data=None)
    name = request.args.get('name', None)
    tasks = Task.query_task(project_id=project_id, folder_id=folder_id, name=name)
    rs = [task.to_json() for task in tasks]
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/options', methods=['GET'])
@exception_handle
def task_options():
    project_id = request.args.get('project_id', None)
    operator = request.args.get('operator', None)
    if project_id is None or operator is None:
        return jsonify(code=1, message='project_id or operator 不能为空', data=None)
    rs = task_service.get_task_options(project_id, operator)
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/<int:task_id>', methods=['PUT'])
@exception_handle
def task_update(task_id):
    data = request.get_json()
    if 'id' not in data:
        data['id'] = task_id
    task = Task(**data)
    task_service.update_task(task)
    return jsonify(code=0, message='ok', data=task.to_json())


@task_bp.route('/tasks/<int:task_id>/command', methods=['GET'])
@exception_handle
def task_cmd(task_id):
    task = Task.get_task(task_id)
    if task:
        command = task.get_task_command()
        return jsonify(code=0, message='ok', data=command)
    else:
        return jsonify(code=404, message='task not found', data=None)


@task_bp.route('/tasks/<int:task_id>/workflow', methods=['GET'])
@exception_handle
def task_workflow(task_id):
    rs = [graph.to_json(include_content=False) for graph in WorkFlow.query_task_workflow(task_id)]
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/instance', methods=['GET'])
@exception_handle
@provide_session
def task_instance_query(size=20, session: Session = None):
    dag_id = request.args.get('dag_id', None)
    task_id = request.args.get('task_id', None)
    execution_date = request.args.get('execution_date', None)
    if dag_id or task_id:
        qry = session.query(TaskInstance)
        if dag_id:
            qry = qry.filter(TaskInstance.dag_id == dag_id)
        if task_id:
            qry = qry.filter(TaskInstance.task_id == task_id)
        if execution_date:
            assert len(str(execution_date)) == 13
            date = dates.from_milliseconds(execution_date)
            qry = qry.filter(TaskInstance.execution_date == date)
        rs = [ti.to_json() for ti in qry.order_by(TaskInstance.execution_date.desc()).limit(size).all()]
        return jsonify(code=0, message='ok', data=rs)
    else:
        return jsonify(code=400, message='dag_id and task_id is null', data=None)


@task_bp.route('/tasks/kill', methods=['POST'])
@exception_handle
@provide_session
def task_instance_kill(session: Session = None):
    data = request.get_json()
    dag_id = data.get('dag_id', None)
    task_id = data.get('task_id', None)
    execution_date = data.get('execution_date', None)
    if dag_id and task_id and execution_date:
        assert len(str(execution_date)) == 13
        date = dates.from_milliseconds(execution_date)
        ti = session.query(TaskInstance).filter(TaskInstance.dag_id == dag_id,
                                                TaskInstance.task_id == task_id,
                                                TaskInstance.execution_date == date).first()
        if ti is not None:
            if ti.state == State.RUNNING:
                ti.is_kill = True
                ti.set_state(State.SHUTDOWN)
                return jsonify(code=0, message='ok', data=ti.to_json())
            else:
                return jsonify(code=400, message='任务不是运行状态', data=ti.to_json())
        else:
            return jsonify(code=404, message='task instance not found', data=None)
    else:
        return jsonify(code=400, message='dag_id or task_id or execution_date is null', data=None)


@task_bp.route('/tasks/<int:task_id>/alerts', methods=['POST'])
@exception_handle
def delete_task_alert(task_id):
    alerts = request.get_json()
    alert_service.save_task_alerts(task_id, alerts)
    return jsonify(code=0, message='ok', data=None)


@task_bp.route('/tasks/<int:task_id>/alerts', methods=['GET'])
@exception_handle
def get_task_alert(task_id):
    condition = int(request.args.get('condition', 0))
    rs = alert_service.get_task_alerts_display(task_id, condition=condition)
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/history_instance', methods=['GET'])
@exception_handle
def get_task_history_instance():
    task_id = request.args.get('task_id', None)
    tis = task_service.get_history_instance(task_id)
    return jsonify(code=0, message='ok', data=tis)


@task_bp.route('/tasks/<int:task_id>/last_instance', methods=['GET'])
@exception_handle
def get_task_last_instance(task_id):
    tis = task_service.get_history_instance(task_id)
    if len(tis) > 0:
        return jsonify(code=0, message='ok', data=tis[0])
    else:
        return jsonify(code=0, message='ok', data=None)


@task_bp.route('/tasks/log', methods=['GET'])
@exception_handle
def get_task_log():
    dag_id = request.args.get('dag_id', None)
    task_id = request.args.get('task_id', None)
    if dag_id and task_id:
        rs = task_service.get_task_log(dag_id, task_id)
        return jsonify(code=0, message='ok', data=rs)
    else:
        return jsonify(code=400, message='hostname is None', data=None)


@task_bp.route('/tasks/log/download', methods=['GET'])
@exception_handle
def download_task_log():
    hostname = request.args.get('hostname', None)
    operator = request.args.get('operator', None)
    start = request.args.get('start', -10000, type=int)
    if hostname:
        return Response(task_service.download_task_log(hostname, operator, start),
                        headers={
                            'Content-Disposition': 'attachment; filename={}.log'.format(hostname)
                        },
                        content_type='application/octet-stream')
    else:
        return jsonify(code=400, message='hostname is None', data=None)


@task_bp.route('/tasks/executor_log/download', methods=['GET'])
@exception_handle
def download_executor_log():
    spark_exec_id = request.args.get('spark_exec_id', None)
    spark_app =  request.args.get('spark_app', None)
    start = request.args.get('start', -10000, type=int)
    if spark_app and spark_exec_id:
        return Response(task_service.download_executor_log(spark_app, spark_exec_id, start),
                        headers={
                            'Content-Disposition': 'attachment; filename={}-{}.log'.format(spark_app, spark_exec_id)
                        },
                        content_type='application/octet-stream')
    else:
        return jsonify(code=400, message='spark_app or spark_exec_id is None', data=None)


@task_bp.route('/tasks/<int:task_id>/blood', methods=['POST'])
@exception_handle
def insert_task_blood(task_id):
    blood = request.get_json()
    Task.update_blood(task_id, blood)
    return jsonify(code=0, message='ok', data=None)


@task_bp.route('/tasks/state_agg', methods=['GET'])
@exception_handle
def task_state_agg():
    start_time = request.args.get('start_time', None, type=int)
    end_time = request.args.get('end_time', None, type=int)
    project_id = request.args.get('project_id', None, type=int)
    if start_time is None:
        log.error("task_state_agg error start_time=%s", start_time, )
        return jsonify(code=400, message="task_state_agg error start_time=None", data=None)
    if end_time is None:
        log.error("task_state_agg error end_time=%s", end_time, )
        return jsonify(code=400, message="task_state_agg error end_time=None", data=None)

    rs = task_service.get_by_start_date(project_id, start_time, end_time)
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/metrics', methods=['GET'])
@exception_handle
def task_metrics():
    start_time = request.args.get('start_time', None, type=int)
    end_time = request.args.get('end_time', None, type=int)
    type = request.args.get('type', None, type=str)
    project_id = request.args.get('project_id', None, type=int)
    if start_time is None:
        log.error("task_state_agg error start_time=%s", start_time, )
        return jsonify(code=400, message="task_state_agg error start_time=None", data=None)
    if end_time is None:
        log.error("task_state_agg error end_time=%s", end_time, )
        return jsonify(code=400, message="task_state_agg error end_time=None", data=None)

    rs = task_service.get_task_metrics(start_time, end_time, type, project_id)
    return jsonify(code=0, message='ok', data=rs)


@task_bp.route('/tasks/proxy', methods=['GET'])
@exception_handle
def task_proxy():
    # base_url = request.base_url
    base_url = conf.get('yarn', 'base_url')
    url = request.args.get('url', None, type=str)
    domain = CommonUtil.get_domain(url)
    s = requests.Session()
    try:
        page = s.get(url)
    except:
        return jsonify(code=404, message='not found', data=None)
    if url.endswith("applications"):
        return Response(page.content, headers={'Content-Type': 'application/jso'})
    if url.endswith("allexecutors"):
        return Response(page.content, headers={'Content-Type': 'application/jso'})
    if url.endswith(".png"):
        return Response(page.content, headers={'Content-Type': 'image/png'})
    if url.endswith(".png"):
        return Response(page.content, headers={'Content-Type': 'image/png'})
    if url.endswith(".css"):
        return Response(page.content, headers={'Content-Type': 'text/css'})
    if url.endswith(".html"):
        return Response(page.content, headers={'Content-Type': 'text/html'})
    if url.endswith(".woff2"):
        return Response(page.content, headers={'Content-Type': 'application/font-woff2'})
    if url.endswith(".js"):
        if url.endswith("executorspage.js"):
            rs = str(page.content, encoding="utf-8")
            path = url[len(domain):]
            suffix = "/".join(path.split("/")[1:3])
            rs = rs.replace("threadDump/?executorId=",
                            "/api/idata/tasks/proxy?url=" + domain + "/executors/threadDump/?executorId=")
            return Response(rs, headers={'Content-Type': 'application/javascript'})
        if url.endswith("utils.js"):
            rs = str(page.content, encoding="utf-8")
            path = url[len(domain):]
            suffix = "/".join(path.split("/")[1:3])
            rs = rs.replace("/api/v1/applications",
                            "/api/idata/tasks/proxy?url=" + domain + "/api/v1/applications")
            rs = rs.replace("/static/",
                            "/api/idata/tasks/proxy?url=" + domain + "/static/")
            return Response(rs, headers={'Content-Type': 'application/javascript'})
        return Response(page.content, headers={'Content-Type': 'application/javascript'})
    if url.endswith("allexecutors"):
        rs = str(page.content, encoding="utf-8")
        rs = rs.replace("http", "/api/idata/tasks/proxy?url=http")
        return rs
    if url.endswith(".json") or url.endswith("config"):
        return Response(page.content)
    soup = BeautifulSoup(page.content, 'lxml')
    elements = []
    elements.append(soup.find_all("a"))
    elements.append(soup.find_all("link"))
    elements.append(soup.find_all("script"))
    elements.append(soup.find_all("img"))
    elements.append(soup.find_all("png"))
    for element in elements:
        for elem in element:
            attr_name = "href"
            if elem.get("src", None) is not None:
                attr_name = "src"

            u = elem.get(attr_name, None)
            if u and u.startswith("/"):
                if "jobhistory" in u:
                    jobhistory_url = conf.get('yarn', 'yarn_history_url')
                    elem[attr_name] = base_url + "?url=" + jobhistory_url + u
                else:
                    elem[attr_name] = base_url + "?url=" + domain + u
            elif u and re.match(CommonUtil.PATTERN, u):
                path = url[len(domain):]
                suffix = "/".join(path.split("/")[1:3])
                elem[attr_name] = base_url + "?url=" + domain + "/" + suffix + "/" + u
            if elem.name and elem.name == "script":
                if elem.string and "attemptsTableData" in elem.string:
                    elem.string = elem.string.replace("href='", "href='/api/idata/tasks/proxy?url=")

    return soup.prettify()


@task_bp.route('/tasks/f_proxy/<address>/proxy/<app_id>/', defaults={'path': ''})
@task_bp.route('/tasks/f_proxy/<address>/proxy/<app_id>/<path:path>', methods=['GET'])
@exception_handle
def task_flink_proxy(address, app_id, path):
    base_url = request.url
    url = "http://" + base_url[base_url.index("/proxy") + 7:]
    page = requests.get(url)
    if url.endswith(".png") or url.endswith(".png?"):
        return Response(page.content, headers={'Content-Type': 'image/png;charset=utf-8'})
    if url.endswith(".svg") or url.endswith(".svg?"):
        return Response(page.content, headers={'Content-Type': 'image/svg+xml;charset=utf-8'})
    if url.endswith(".css") or url.endswith(".css?"):
        return Response(page.content, headers={'Content-Type': 'text/css;charset=utf-8'})
    if url.endswith(".woff2") or url.endswith(".woff2?"):
        return Response(page.content, headers={'Content-Type': 'application/font-woff2;charset=utf-8'})
    if url.endswith(".js") or url.endswith(".js?"):
        return Response(page.content, headers={'Content-Type': 'application/javascript;charset=utf-8'})
    return Response(page.content, headers={'Content-Type': 'application/json;charset=utf-8'})


@task_bp.route('/tasks/f_proxy/<address>/proxy/<app_id>/flink', methods=['GET'])
@exception_handle
def task_flink_proxy2(address, app_id):
    address = "http://" + address
    home_url = "http://" + app_id
    page = requests.get(home_url)
    if page and page.content and "Apache Flink Web Dashboard" not in page.text:
        return redirect("/api/idata/tasks/proxy?url={}/proxy/{}".format(address, app_id))
    soup = BeautifulSoup(page.content, 'lxml')
    return soup.prettify()


@task_bp.route('/pods', methods=['GET'])
@exception_handle
def list_pod():
    status = request.args.get('status', None)
    app = request.args.get('app', None)
    return jsonify(code=0, message='ok', data=task_service.list_namespaced_pod('idata', status=status, app=app))


@task_bp.route('/tasks/datafactory/plugin', methods=['GET'])
@exception_handle
def list_datafactory_plugin():
    t_type = request.args.get('type', None)
    if int(t_type) == 1:
        reader_options = [{'value': EtlUtil.SOURCE_TYPE_KAFKA, 'label': 'KAFKA', 'nodeType': 'reader'}]
        writer_options = [{'value': EtlUtil.SINK_TYPE_KAFKA, 'label': 'KAFKA', 'nodeType': 'writer'},
                          {'value': EtlUtil.SINK_TYPE_HIVE, 'label': 'HIVE', 'nodeType': 'writer'},
                          {'value': EtlUtil.SINK_TYPE_ES, 'label': 'ES', 'nodeType': 'writer'},
                          {'value': EtlUtil.SINK_TYPE_DORIS, 'label': 'DORIS', 'nodeType': 'writer'}]
        filter_options = [{'value': 1, 'label': 'origin', 'nodeType': 'filter'},
                          {'value': 2, 'label': 'custom', 'nodeType': 'filter'},
                          {'value': 3, 'label': 'sql', 'nodeType': 'filter'}]
    else:
        reader_options = [{'value': EtlUtil.SOURCE_TYPE_HIVE, 'label': 'HIVE', 'nodeType': 'reader'},
                          {'value': EtlUtil.SOURCE_TYPE_JDBC, 'label': 'JDBC', 'nodeType': 'reader'}]
        writer_options = [{'value': EtlUtil.SINK_TYPE_HIVE, 'label': 'HIVE', 'nodeType': 'writer'},
                          {'value': EtlUtil.SINK_TYPE_JDBC, 'label': 'JDBC', 'nodeType': 'writer'}]
        filter_options = [{'value': 3, 'label': 'sql', 'nodeType': 'filter'}]

    result = {
        "reader_options": reader_options,
        "writer_options": writer_options,
        "filter_options": filter_options
    }

    return jsonify(code=0, message='ok', data=result)


@task_bp.route('/tasks/datafactory/fieldtypes', methods=['GET'])
@exception_handle
def get_fieldTypes():
    options = [{'value': 'binary', 'label': 'binary'}, {'value': 'boolean', 'label': 'boolean'},
               {'value': 'byte', 'label': 'byte'}, {'value': 'date', 'label': 'date'},
               {'value': 'double', 'label': 'double'}, {'value': 'float', 'label': 'float'},
               {'value': 'int', 'label': 'int'}, {'value': 'long', 'label': 'long'},
               {'value': 'short', 'label': 'short'}, {'value': 'string', 'label': 'string'},
               {'value': 'timestamp', 'label': 'timestamp'}, {'value': 'null', 'label': 'null'},
               ]

    return jsonify(code=0, message='ok', data=options)


@task_bp.route('/tasks/datafactory/run_resource', methods=['GET'])
@exception_handle
def get_run_resource():
    driver_options = [
        {'value': '1核-1g', 'label': '1核-1g', 'default': True}, {'value': '1核-2g', 'label': '1核-2g', 'default': False},
        {'value': '2核-1g', 'label': '2核-1g', 'default': False}, {'value': '2核-2g', 'label': '2核-2g', 'default': False},
        {'value': '2核-4g', 'label': '2核-4g', 'default': False}, {'value': '2核-6g', 'label': '2核-6g', 'default': False},
        {'value': '3核-1g', 'label': '3核-1g', 'default': False}, {'value': '3核-3g', 'label': '3核-3g', 'default': False},
        {'value': '3核-6g', 'label': '3核-6g', 'default': False}, {'value': '4核-1g', 'label': '4核-1g', 'default': False},
        {'value': '4核-2g', 'label': '4核-2g', 'default': False}, {'value': '4核-4g', 'label': '4核-4g', 'default': False},
    ]

    executor_options = [
        {'value': '1核-1g', 'label': '1核-1g', 'default': True}, {'value': '1核-2g', 'label': '1核-2g', 'default': False},
        {'value': '1核-4g', 'label': '1核-4g', 'default': False},
        {'value': '2核-1g', 'label': '2核-1g', 'default': False}, {'value': '2核-2g', 'label': '2核-2g', 'default': False},
        {'value': '2核-4g', 'label': '2核-4g', 'default': False}, {'value': '2核-6g', 'label': '2核-6g', 'default': False},
        {'value': '2核-8g', 'label': '2核-8g', 'default': False},
        {'value': '2核-12g', 'label': '2核-12g', 'default': False},
        {'value': '2核-14g', 'label': '2核-14g', 'default': False},
        {'value': '3核-1g', 'label': '3核-1g', 'default': False}, {'value': '3核-3g', 'label': '3核-3g', 'default': False},
        {'value': '3核-6g', 'label': '3核-6g', 'default': False}, {'value': '3核-9g', 'label': '3核-9g', 'default': False},
        {'value': '3核-12g', 'label': '3核-12g', 'default': False},
        {'value': '4核-1g', 'label': '4核-1g', 'default': False},
        {'value': '4核-2g', 'label': '4核-2g', 'default': False}, {'value': '4核-4g', 'label': '4核-4g', 'default': False},
        {'value': '4核-6g', 'label': '4核-6g', 'default': False}, {'value': '4核-8g', 'label': '4核-8g', 'default': False},
        {'value': '4核-12g', 'label': '4核-12g', 'default': False},
        {'value': '5核-1g', 'label': '5核-1g', 'default': False},
        {'value': '5核-2g', 'label': '5核-2g', 'default': False}, {'value': '5核-3g', 'label': '5核-3g', 'default': False},
        {'value': '5核-4g', 'label': '5核-4g', 'default': False}, {'value': '5核-5g', 'label': '5核-5g', 'default': False},
        {'value': '5核-10g', 'label': '5核-10g', 'default': False},
        {'value': '6核-1g', 'label': '6核-1g', 'default': False}, {'value': '6核-2g', 'label': '6核-2g', 'default': False},
        {'value': '6核-3g', 'label': '6核-3g', 'default': False}, {'value': '6核-6g', 'label': '6核-6g', 'default': False},
        {'value': '6核-12g', 'label': '6核-12g', 'default': False},
        {'value': '8核-1g', 'label': '8核-1g', 'default': False}, {'value': '8核-2g', 'label': '8核-2g', 'default': False},
        {'value': '8核-4g', 'label': '8核-4g', 'default': False}, {'value': '8核-8g', 'label': '8核-8g', 'default': False},
        {'value': '9核-1g', 'label': '9核-1g', 'default': False}, {'value': '9核-3g', 'label': '9核-3g', 'default': False},
        {'value': '9核-9g', 'label': '9核-9g', 'default': False},
        {'value': '10核-1g', 'label': '10核-1g', 'default': False},
        {'value': '10核-2g', 'label': '10核-2g', 'default': False},
        {'value': '10核-5g', 'label': '10核-5g', 'default': False},
        {'value': '10核-10g', 'label': '10核-10g', 'default': False},
    ]

    options = {
        "driver_options": driver_options,
        "executor_options": executor_options
    }

    return jsonify(code=0, message='ok', data=options)


@task_bp.route('/tasks/datafactory/metrics', methods=['GET'])
@exception_handle
def get_task_metrics():
    t_id = request.args.get('task_id', None)
    t_name = request.args.get('task_name', None)
    start_time = request.args.get('start_time', None, type=int)
    end_time = request.args.get('end_time', None, type=int)
    rs = task_service.get_datafactory_metrics(start_time, end_time, t_id, t_name)
    return jsonify(code=0, message='ok', data=rs)
